package com.lazy.bench;

import java.io.IOException;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;

import com.lazy.tcp.AysncChannelFutureListener;
import com.lazy.tcp.TelnetClient;
import com.lazy.tcp.TelnetClientHandler;


public class BenchMain {

    private static Logger logger = Logger.getLogger("BenchMain");
    /** threadid<-> channel */
    private static Map<Long, Channel> channelMap = new ConcurrentHashMap<Long, Channel>();
    
    private ExecutorService executor = null;
    TelnetClient client;
    TelnetClientHandler clientHandler;
    
    
    static AysncChannelFutureListener clientListener = new AysncChannelFutureListener("post");
    
    private AtomicInteger nSend = new AtomicInteger();
    private AtomicInteger nErrorSend = new AtomicInteger();
    private Date firstSendDate = null;
    private Date lastSendDate = null;
    
    public BenchMain() {
        AppCfg cfg = AppCfg.getCfg();
        executor = Executors.newFixedThreadPool(cfg.clientMaxThreads);
        
        clientHandler = new TelnetClientHandler();

        client = new TelnetClient(cfg.serverIp, cfg.serverPort, clientHandler);
    }
    
    public void run () {
        final AppCfg cfg = AppCfg.getCfg();
        logger.info("runint with " + cfg.clientMaxThreads + " threads and send data count " + cfg.clientTotalPack);

        Thread task = new Thread() {
            @Override
            public void run() {
                try {
                    ChannelFuture lastWriteFuture = send(client);
                    if(lastWriteFuture == null || !lastWriteFuture.isSuccess()) {
                        logger.info(Thread.currentThread().getName() + " send with null ChannelFuture " + new Date());
                        nErrorSend.addAndGet(1);
                    }
                    else {
                        nSend.addAndGet(1);
                        lastSendDate = new Date();
                    }

                } catch (Exception e) {
                    nErrorSend.addAndGet(1);
                    e.printStackTrace();
                    return;
                }
            }
        };

        firstSendDate = new Date();
        logger.info(Thread.currentThread().getName() + "=========begin send time======" + firstSendDate);
        for (int i=0; i < cfg.clientTotalPack; i++) {
            // sleep a give time
            if (cfg.clientSendInterval > 0) {
                try{
                    Thread.sleep(cfg.clientSendInterval);
                }catch(InterruptedException e){
                    e.printStackTrace();
                }
            }
            executor.submit(task);
        }
    }
    
    private ChannelFuture send(TelnetClient client) throws IOException {
        AppCfg cfg = AppCfg.getCfg();
        String content = cfg.clientPackContent;

        long threadid = Thread.currentThread().getId();
        Channel channel = channelMap.get(threadid);
        if (channel == null || !channel.isConnected()) {
            channel = client.connect();
            if (channel != null) {
                channelMap.put(threadid, channel);
            }
        }
        ChannelFuture lastWriteFuture = null;

        if (channel != null) {
            lastWriteFuture= client.post(channel, content);
            logger.info("send string:[" + nSend.get() + "]" + content);

            // Wait until all messages are flushed before closing the channel.
            if (lastWriteFuture != null) {
                // lastWriteFuture.awaitUninterruptibly();
                lastWriteFuture.addListener(clientListener);
            }
        }
        return lastWriteFuture;
    }


    /**
     * 检测客户端的运行状况
     */
    public void monitor() {
        new Thread() {
            @Override
            public void run() {
                while (true) {
                    try {
                        sleep(500);

                        if (nSend.get() + nErrorSend.get() >= AppCfg.getCfg().clientTotalPack) {
                            for (Channel channel:channelMap.values()) {
                                client.disconnect(channel);
                            }
                            client.close();
                            executor.shutdown();
                            System.out.println("[" + firstSendDate + ", " + lastSendDate + "]");
                            System.out.println("========End=========");
                            System.out.println("live:"+ channelMap.size() + " sended:" + nSend.get()
                                    + " error send:" + nErrorSend.get()
                                    + " received:" + TelnetClientHandler.getReceivePacketCount());
                            // return;
                        }
                        else {
                            System.out.println("live:"+ channelMap.size() + " sended:" + nSend.get()
                                    + " error send:" + nErrorSend.get()
                                    + " received:" + TelnetClientHandler.getReceivePacketCount());
                        }

                    } catch (InterruptedException e) {
     
                    }
                }
            }
        }.start();
    }

    public static void main(String[] args) throws Exception {

        BenchMain clientTest = new BenchMain();
        clientTest.run();
        clientTest.monitor();
    }
}
